package com.deep.flink;

import javafx.stage.Stage;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

public class NFAExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 获取訂單事件 user1
        KeyedStream<LogEvent, String> loginEventStringKeyedStream = env.fromElements(
                new LogEvent("user_1", "192.168.0.1", "fail", 2000L),
                new LogEvent("user_1", "192.168.0.2", "fail", 3000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 4000L),
                new LogEvent("user_1", "192.168.23.10", "fail", 5000L),
                new LogEvent("user_2", "192.168.1.29", "success", 6000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 7000L),
                new LogEvent("user_2", "192.168.1.29", "fail", 8000L)

        ).keyBy(event -> event.userId);

        // 数据按照顺序 依次输入
        SingleOutputStreamOperator<String> warningStream = loginEventStringKeyedStream.flatMap(new StateMachineMapper());

        warningStream.print();

        env.execute();

    }

    public static class StateMachineMapper extends RichFlatMapFunction<LogEvent, String> {
        ValueState<State> currentState;

        @Override
        public void open(Configuration parameters) throws Exception {
            currentState = getRuntimeContext().getState(new ValueStateDescriptor<State>(
                    "state", State.class
            ));
        }

        @Override
        public void flatMap(LogEvent logEvent, Collector<String> collector) throws Exception {
            // 如果状态为空，进行初始化
            State state = currentState.value();
            if (state == null) {
                state = State.Initial;
            }

            //
            State nextState = state.transition(logEvent.eventType);
            // 判断当前特殊情况
            if(nextState == State.Matched) {
                // 检测到了匹配，输出报警信息
                collector.collect(logEvent.userId + " 连续3次登录失败");
            } else if (nextState == State.Terminal) {
                // 更新为初始化
                currentState.update(State.Initial);
            } else {
                currentState.update(nextState);
            }

        }
    }

    public enum State {
        Terminal,   // 匹配失败，终止

        Matched,   // 匹配成功
        S2(new Transition("fail", Matched), new Transition("success", Terminal)),
        S1(new Transition("fail", S2), new Transition("success", Terminal)),
        Initial(new Transition("fail", S1), new Transition("success", Terminal));


        private Transition[] transitions ;
        State(Transition... transitions) {
            this.transitions = transitions;
        }

        public State transition(String eventType) {
            for(Transition transition: transitions) {
                if (transition.eventType.equals(eventType)) {
                    // 转移
                    return transition.getTargetState();
                }
            }
            // 回到初始状态
            return Initial;
        }
    }

    // 定义 一个状态转移类， 包含当前引起状态转移的事件类型，以及转移的目标状态

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Transition {
        private String eventType;
        private State targetState;

    }
}
